package com.cetc.sdp.kmga.cs.stream;

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @description:
 * @author： DengQiang
 * @date: 2017/8/23 10:18
 */
public abstract class AbstractBroadcastWrapper<T> {
    private volatile Broadcast<T> var;

    private long lastUpdateTime = 0L;

    private int interval;

    private boolean updateByFixRate = false;

    private AtomicBoolean refresh = new AtomicBoolean(false);

    private JavaSparkContext jsc;

    public AbstractBroadcastWrapper(JavaSparkContext jsc) {
        this.interval = 3600000;
        this.jsc = jsc;
    }

    public Broadcast<T> updateAndGet() {
        if (var == null) {
            System.out.println("Init broadcast");
            this.var = broadcast(jsc);
            lastUpdateTime = System.currentTimeMillis();
        } else if (refresh.get()) {
            System.out.println("Unpersist old broadcast");
            System.out.println("refresh broadcast");
            var.unpersist();
            var = broadcast(jsc);
            refresh.compareAndSet(true, false);
            lastUpdateTime = System.currentTimeMillis();
        } else if (updateByFixRate) {
            long diff = System.currentTimeMillis() - lastUpdateTime;
            if (diff > interval) {
                System.out.println("Unpersist old broadcast");
                System.out.println("refresh broadcast by fix rate");
                var.unpersist();
                reload();
                var = broadcast(jsc);
                lastUpdateTime = System.currentTimeMillis();
            }
        }
        return var;
    }

    /**
     * 刷新广播变量
     */
    public void refreshBroadcast() {
        refresh.compareAndSet(false, true);
    }

    protected JavaSparkContext getJsc() {
        return jsc;
    }

    public int getInterval() {
        return interval;
    }

    public void setInterval(int interval) {
        this.interval = interval;
    }

    public void setUpdateByFixRate(boolean updateByFixRate) {
        this.updateByFixRate = updateByFixRate;
    }

    /**
     * 重新读取数据
     */
    protected void reload() {}

    /**
     * 广播
     * @param jsc
     * @return
     */
    protected abstract Broadcast<T> broadcast(JavaSparkContext jsc);
}
